package com.atguigu.app.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.DWDStaticConstants;
import com.atguigu.mapper.DimAsyncJoinFunction;
import com.atguigu.utils.DimUtil;
import com.atguigu.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author:GaoFei
 * @Description:
 * @Date:Created in 18:11
 * @Modified By:
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimAsyncJoinFunction<T> {
    private Connection connection;
    private ThreadPoolExecutor poolExecutor;
    private final String tableName;
    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(DWDStaticConstants.HBASE_PHOENIX_DRIVER);
        connection = DriverManager.getConnection(DWDStaticConstants.HBASE_PHOENIX_SERVER);
        // get one threadPoolExecutor
        poolExecutor = ThreadPoolUtil.getThreadPool();
    }

    @Override
    public void asyncInvoke(T t, ResultFuture<T> resultFuture) throws Exception {
        poolExecutor.submit(() -> {
            try {
                JSONObject jsonObject = DimUtil.obtainJson(connection, tableName, getId(t));
                join(jsonObject,t);
                resultFuture.complete(Collections.singletonList(t));
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }



    // 处理异步请求中的超时请求
    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {

    }
}
